package com.lefu.risk.storm.bolt;

import java.io.File;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;




public class ReportBolt extends BaseRichBolt{

	private static final long serialVersionUID = 1L;
	
	private Map<String,Long> _countMap = null;

	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this._countMap = new HashMap<>();
	}

	@Override
	public void execute(Tuple input) {
		String word = input.getStringByField("word");
		Long count = input.getLongByField("count");
		this._countMap.put(word, count);
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		
	}
	
	@SuppressWarnings("resource")
	@Override
	public void cleanup() {
		List<String> keys = new ArrayList<>(this._countMap.keySet());
		Collections.sort(keys);
		PrintWriter print = null;
		try {
			print = new PrintWriter(new File("F:\\field_wordcount.txt"));
			for (String key : keys) {
				print.println(key+" : "+_countMap.get(key));
			}
		} catch (IOException e) {
			e.printStackTrace();
		}finally{
			if(print != null){
				try {
					print.close();
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
		
		for (String key : keys) {
			System.out.println(key+" : "+ this._countMap.get(key)); 
		}
		
		System.out.println("----cleanup----");
		
	}

}
